﻿/*  ----------------------------------------------------------------------------
 *  SysServer
 *  ----------------------------------------------------------------------------
 *  File:       IOCPManager.cs
 *  Creater:     SEEN\MistSeen
 *  ----------------------------------------------------------------------------
 *  $Date: 2010-09-06 10:38:08 +0000 (Mon, 06 Sep 2010) $
 *  $Rev: 16 $
 *  $Author: ScorDeity $
 *  $URL: http://sysserver.googlecode.com/svn/trunk/src/SysServer/IOCPManager/IOCPManager.cs $
 *  $Id: IOCPManager.cs 16 2010-09-06 10:38:08Z ScorDeity $
 */

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;


namespace SysServer
{
    using System.Threading;
    using System.Collections;
    /// <summary>
    /// IOCP Handle used by Threads to wait on IOCP events
    /// </summary>
    public class IOCPHandle
    {
        private AutoResetEvent _AutoEvent = new AutoResetEvent(false);
        private bool _WaitingOnIOCP = false;
        /// <summary>
        /// Identifies whether a Thread using this IOCP Handle is waiting for
        /// the availability of an object in the Managed IOCP queue. Used by 
        /// Managed IOCP in its default 'ChooseThread' policy
        /// </summary>
        public bool WaitingOnIOCP
        {
            get { return this._WaitingOnIOCP; }
        }
        private bool _FirstRun = true;
        private IOCPManager _Manager;
        private Thread _OwningThread;
        /// <summary>
        /// Retrieves the owning thread for this IOCP Handle instance. Used by 
        /// Managed IOCP in its default 'ChooseThread' policy
        /// </summary>
        public Thread OwningThread
        {
            get { return this._OwningThread; }
        }
        internal int _ActiveState = 0;
        /// <summary>
        /// Identifies whether a Thread using this IOCP Handle has went past the 'Wait'
        /// state. This means that the thread is actively operating on a queued object
        /// of a Managed IOCP instance
        /// </summary>
        public int ActiveState
        {
            get { return this._ActiveState; }
        }
        /// <summary>
        /// Identifies if the current instance of IOCPHandle is holding
        /// a valid Managed IOCP instance. Valid could return null if the
        /// Managed IOCP instance to which this IOCPHandle instance belongs
        /// to is closed
        /// </summary>
        public bool Valid
        {
            get { return this._Manager == null ? false : true; }
        }
        /// <summary>
        /// Internal constructor used by Managed IOCP to create an instance of 
        /// IOCP Handle. This is called when a new thread registers with a
        /// Managed IOCP instance
        /// </summary>
        /// <param name="mIOCP">Managed IOCP instance</param>
        /// <param name="owningThread">Thread that is trying to register with the
        /// Managed IOCP instance identified by 'mIOCP' param</param>
        internal IOCPHandle(IOCPManager manager, Thread owningThread)
        {
            this._Manager = manager;
            this._OwningThread = owningThread;
        }
        /// <summary>
        /// Used internally by Managed IOCP instance to notify the thread waiting on this
        /// IOCP Handle about availability of a queued object
        /// </summary>
        internal bool AutoEventSet()
        {
            return this._AutoEvent.Set();
        }
        /// <summary>
        /// Used internally by Managed IOCPinstance to close the AutoResetEvent handle
        /// help by this instance of IOCPHandle
        /// </summary>
        internal void AutoEventClose()
        {
            this._AutoEvent.Close();
        }
        /// <summary>
        /// Thread calling this method will be suspended untill a new object
        /// is available in the Managed IOCP queue.If the current thread has
        /// unregistered with the ManagedIOCP instance which is held by this
        /// IOCPHandle instance, then an exception is thrown.
        /// </summary>
        /// <returns>Object from the ManagedIOCP queue</returns>
        public object Wait()
        {
            if (this._Manager != null)
            {
                try
                {
                    if (this._ActiveState == 1)
                    {
                        this._Manager.DecrementActiveThreads();
                        this._ActiveState = 0;
                    }
                    else if (this._ActiveState == 2)
                    {
                        this._ActiveState = 0;
                    }

                    if (this._FirstRun == true)
                    {
                        this._Manager.EnqueueHandle(this);
                        this._FirstRun = false;
                    }
                    object entity = null;
                    do
                    {
                        if (this._Manager.IsRuning == true)
                        {
                            if (this._Manager.IncrementActiveThreads() == true)
                            {
                                bool empty = false;
                                entity = this._Manager.GetNextEntity(ref empty);
                                if (empty == true)
                                {
                                    this._Manager.DecrementActiveThreads();
                                    this._WaitingOnIOCP = true;
                                    this._AutoEvent.WaitOne();
                                    this._WaitingOnIOCP = false;
                                }
                                else
                                {
                                    break;
                                }
                            }
                            else
                            {
                                this._WaitingOnIOCP = true;
                                this._AutoEvent.WaitOne();
                                this._WaitingOnIOCP = false;
                            }
                        }
                        else
                        {
                            this._WaitingOnIOCP = true;
                            this._AutoEvent.WaitOne();
                            this._WaitingOnIOCP = false;
                        }

                    } while (true);

                    this._ActiveState = 1;

                    this._Manager.EnqueueHandle(this);

                    return entity;
                }
                catch (Exception ex)
                {
                    if (this._Manager == null)
                        throw new Exception("Invalid ManagedIOCP instance." +
                            " Check whether you have unregistered the current thread from the" +
                            " ManagedIOCP instance held by this IOCPHandle .or. ManagedIOCP" +
                            " instance is closed.");
                    else
                    {
                        throw ex;
                    }
                }

            }
            else
            {
                throw new Exception("Invalid ManagedIOCP instance." +
                                    " Check whether you have unregistered the current thread from the" +
                                    " ManagedIOCP instance held by this IOCPHandle.or. ManagedIOCP" +
                                    " instance is closed.");
            }
        }
        /// <summary>
        /// Thread calling this method will be suspended untill a new object
        /// is available in the Managed IOCP queue.If the current thread has
        /// unregistered with the ManagedIOCP instance which is held by this
        /// IOCPHandle instance, then an exception is thrown.
        /// If specified timeout expired before the current thread could grab 
        /// an object from the associated ManagedIOCP queue then the function
        /// returne to caller with null object and with timeOut parameter 
        /// value set to true.
        /// </summary>
        /// <param name="millisecondsTimeout">Milliseconds to wait for object arrival</param>
        /// <param name="timeOut">Set to true if the Wait timed-out else false</param>
        /// <returns></returns>
        public object Wait(int millisecondsTimeout, ref bool timeOut)
        {
            if (this._Manager != null)
            {
                try
                {
                    if (this._ActiveState == 1)
                    {
                        this._Manager.DecrementActiveThreads();
                        this._ActiveState = 0;
                    }
                    else if (this._ActiveState == 2)
                    {
                        this._ActiveState = 0;
                    }

                    if (this._FirstRun == true)
                    {
                        this._Manager.EnqueueHandle(this);
                        this._FirstRun = false;
                    }
                    object entity = null;
                    do
                    {
                        if (this._Manager.IsRuning == true)
                        {
                            if (this._Manager.IncrementActiveThreads() == true)
                            {
                                bool empty = false;
                                entity = this._Manager.GetNextEntity(ref empty);
                                if (empty == true)
                                {
                                    this._Manager.DecrementActiveThreads();
                                    this._WaitingOnIOCP = true;
                                    if (this._AutoEvent.WaitOne(millisecondsTimeout, false) == false)
                                    {
                                        timeOut = true;
                                        this._WaitingOnIOCP = false;
                                        break;
                                    }
                                    else
                                    {
                                        this._WaitingOnIOCP = false;
                                    }
                                }
                                else
                                {
                                    break;
                                }
                            }
                            else
                            {
                                this._WaitingOnIOCP = true;
                                if (this._AutoEvent.WaitOne(millisecondsTimeout, false) == false)
                                {
                                    timeOut = true;
                                    this._WaitingOnIOCP = false;
                                    break;
                                }
                                else
                                {
                                    this._WaitingOnIOCP = false;
                                }
                            }
                        }
                        else
                        {
                            this._WaitingOnIOCP = true;
                            if (this._AutoEvent.WaitOne(millisecondsTimeout, false) == false)
                            {
                                timeOut = true;
                                this._WaitingOnIOCP = false;
                                break;
                            }
                            else
                            {
                                this._WaitingOnIOCP = false;
                            }
                        }

                    } while (true);

                    this._ActiveState = 1;

                    this._Manager.EnqueueHandle(this);

                    return entity;
                }
                catch (Exception ex)
                {
                    if (this._Manager == null)
                        throw new Exception("Invalid ManagedIOCP instance." +
                            " Check whether you have unregistered the current thread from the" +
                            " ManagedIOCP instance held by this IOCPHandle .or. ManagedIOCP" +
                            " instance is closed.");
                    else
                    {
                        throw ex;
                    }
                }

            }
            else
            {
                throw new Exception("Invalid ManagedIOCP instance." +
                                    " Check whether you have unregistered the current thread from the" +
                                    " ManagedIOCP instance held by this IOCPHandle.or. ManagedIOCP" +
                                    " instance is closed.");
            }
        }

        internal void InvalidateOwningIOCPManager()
        {
            this._Manager = null;
        }
    }

    /// <summary>
    /// Provides an all managed implementation of Native Microsoft Windows 
    /// IOCP feature. This class does not provide IOCP based on I/O handles.
    /// Rather it provides waitable object queuing features of Native IOCP using 
    /// all managed .Net threading and synchronization primitives. 
    /// 
    /// This class provides the following features...
    /// 
    /// 1. Multiple Managed IOCP instances per process
    /// 2. Registration of multiple threads per Managed IOCP instance
    /// 3. Dispatching System.Object types to a queue maintained by each 
    ///		Managed IOCP instance
    /// 4. Waitable multi-thread safe retrieval of objects from the Managed IOCP
    ///		instance queue by all the threads registered for that particular
    ///		Managed IOCP instance
    ///	5. Ability to restrict number of concurrent active threads processing the
    ///		queued objects related to a particular Managed IOCP instance
    ///	6. Policy based replaceable/customizable approach for choosing a 
    ///		registered thread to process next available queued object
    ///	7. Pause the Managed IOCP processing. Internally pauses processing of queued objects
    ///		by registered threads. Also by default disallows enqueuing new objects.
    ///	8. Run the Managed IOCP instance. Internally re-starts the processing of 
    ///		queued objects by registered threads. Also allows enqueuing new objects
    ///	9. Modify the concurrent threads at runtime
    ///	10. Provides easy accesibility to Managed IOCP instance runtime properties like...
    ///		10.1. Number of active concurrent threads
    ///		10.2. Number of objects left in queue
    ///		10.3. Running status
    ///		10.4. Number of allowed concurrent threads
    ///	11. Safe and controlled closing of an Managed IOCP instance. Read the code 
    ///			comments/documentation on ManagedIOCP.Close() method
    /// </summary>
    public class IOCPManager
    {
        private ArrayList _RegIOCPHandles = ArrayList.Synchronized(new ArrayList(5));
        private Hashtable _RegThreads = Hashtable.Synchronized(new Hashtable(5));
        private PoolableQueue _EntityQueue = new PoolableQueue(10000);
        private PoolableQueue _IOCPHanderQueue = new PoolableQueue(100);

        private int _ActiveThreads = 0;

        public int ActiveThreads
        {
            get
            {
                return this._ActiveThreads;
            }
        }
        private int _MaxThreads = 0;

        public int MaxThreads
        {
            get { return this._MaxThreads; }
            set { this._MaxThreads = value; }
        }
        private bool _IsRuning = false;

        public bool IsRuning
        {
            get
            {
                return this._IsRuning;
            }
        }
        private bool _CanEnqueueOnPause = false;

        public bool CanEnqueueOnPause
        {
            get
            {
                return this._CanEnqueueOnPause;
            }
            set
            {
                this._CanEnqueueOnPause = value;
            }
        }

        public long EntityQueueCount
        {
            get { return this._EntityQueue.Count; }
        }

        public long RegisteredThreadsCount
        {
            get { return this._RegThreads.Count; }
        }
        /// <summary>
        /// Public constructor for ManagedIOCP class. Defaults to 1 
        /// concurrent thread that can actively process objects queued to this 
        /// Managed IOCP instance.
        /// 
        /// Note:
        /// -----
        /// We are using Monitor on the ManagedIOCP instance to sync access to 
        /// Register, UnRegister & Close methods. As these methods use enumerators
        /// that fail during multi-threaded operations
        /// </summary>
        public IOCPManager()
            : this(1)
        {

        }
        /// <summary>
        /// Public constructor for ManagedIOCP class
        /// </summary>
        /// <param name="concurrentThreads">Number of concurrent threads that can 
        /// actively process objects queued to this Managed IOCP instance</param>
        public IOCPManager(int currentThreads)
        {
            this._MaxThreads = currentThreads;
        }
        /// <summary>
        /// Registers the calling thread with this instance of ManagedIOCP
        /// </summary>
        /// <returns>IOCPHandle class instance that can be used by this thread
        /// to retrieve the next available object from this ManagedIOCP instance queue.
        /// </returns>

        public IOCPHandle Register()
        {
            IOCPHandle iocpHandle = null;
            Thread th = Thread.CurrentThread;
            try
            {
                Monitor.Enter(this);
                if (this._RegThreads.ContainsKey(th))
                {
                    iocpHandle = this._RegThreads[th] as IOCPHandle;
                }
                else
                {
                    iocpHandle = new IOCPHandle(this, th);
                    this._RegThreads.Add(th, iocpHandle);
                    this._RegIOCPHandles.Add(iocpHandle);
                }
            }
            catch (Exception)
            {
                throw;
            }
            finally
            {
                Monitor.Exit(this);
            }
            return iocpHandle;
        }
        /// <summary>
        /// UnRegisters the calling thread with this instance of ManagedIOCP. After this call
        /// the thread will not be able to wait on its IOCPHandle for receiving any 
        /// objects queued to this instance of ManagedIOCP. The calling thread can re-register
        /// with this instance of ManagedIOCP and receive objects queued to this instance
        /// of ManagedIOCP
        /// </summary>
        public void UnRegister()
        {
            IOCPHandle iocpHandle = null;
            Thread th = Thread.CurrentThread;
            try
            {
                Monitor.Enter(this);
                if (this._RegThreads.ContainsKey(th))
                {
                    iocpHandle = this._RegThreads[th] as IOCPHandle;
                    this._RegThreads.Remove(th);
                    this._RegIOCPHandles.Remove(iocpHandle);
                    iocpHandle.InvalidateOwningIOCPManager();
                    iocpHandle.AutoEventSet();

                    if (iocpHandle.ActiveState == 1)
                    {
                        DecrementActiveThreads();
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                Monitor.Exit(this);
            }
        }

        public void Close()
        {
            try
            {
                Monitor.Enter(this);
                this._IsRuning = false;

                foreach (IOCPHandle handle in this._RegIOCPHandles)
                {
                    handle.InvalidateOwningIOCPManager();
                    handle.AutoEventSet();
                    handle.AutoEventClose();
                }

                this._RegThreads.Clear();
                this._RegIOCPHandles.Clear();
                this._EntityQueue.Clear();

                this._ActiveThreads = 0;
                this._IsRuning = false;
            }
            catch (Exception)
            {
                throw;
            }
            finally
            {
                Monitor.Exit(this);
            }
        }
        /// <summary>
        /// Enqueues a new object ono the queue maintained by this instance 
        /// of ManagedIOCP
        /// </summary>
        /// <param name="obj"></param>
        public void Dispatch(object value)
        {
            int count = this._RegIOCPHandles.Count;
            if (count > 0)
            {
                if ((this._IsRuning == true) || (this._CanEnqueueOnPause == true))
                {
                    this._EntityQueue.Enqueue(value);
                }
                else
                {
                    throw new Exception("Cannot dispatch objects. Currently in pause mode");
                }
                WakeupNextThread();
            }
            else
            {
                throw new Exception("No threads available to service this dispatch request");
            }
        }
        /// <summary>
        /// Pauses the processing of queued objects of this ManagedIOCP instance 
        /// by the threads registered with this instance of ManagedIOCP
        /// </summary>
        public void Pause()
        {
            this._IsRuning = false;
        }
        /// <summary>
        /// Start the processing of queued objects of this ManagedIOCP instance 
        /// by the threads registered with this instance of ManagedIOCP.
        /// This is the default mode for any newly created instance of ManagedIOCP
        /// </summary>
        public void Run()
        {
            this._IsRuning = true;
            this.WakeupNextThread();
        }

        internal bool IncrementActiveThreads()
        {
            bool incremented = true;
            do
            {
                int curActThreads = this._ActiveThreads;
                int newActThreads = curActThreads + 1;
                if (newActThreads <= this._MaxThreads)
                {
                    if (Interlocked.CompareExchange(ref this._ActiveThreads, newActThreads, curActThreads) == curActThreads)
                    {
                        break;
                    }
                }
                else
                {
                    incremented = false;
                    break;
                }


            } while (true);

            return incremented;
        }

        internal void DecrementActiveThreads()
        {
            if (this._ActiveThreads > 0)
            {
                Interlocked.Decrement(ref this._ActiveThreads);
            }
        }

        internal object GetNextEntity(ref bool empty)
        {
            object obj = null;
            if (this._EntityQueue.Count > 0)
            {
                try
                {
                    obj = this._EntityQueue.Dequeue(ref empty);

                }
                catch
                {
                    empty = true;
                }
            }
            else
            {
                empty = true;
            }
            return obj;
        }

        internal void EnqueueHandle(IOCPHandle handle)
        {
            this._IOCPHanderQueue.Enqueue(handle);
        }

        private void WakeupNextThread()
        {
            bool empty = false;
            while (true)
            {
                IOCPHandle handle = this._IOCPHanderQueue.Dequeue(ref empty) as IOCPHandle;
                if ((empty == false) && (handle != null))
                {
                    if (handle.WaitingOnIOCP == true)
                    {
                        handle.AutoEventSet();
                        break;
                    }
                    else
                    {
                        if (handle.OwningThread.ThreadState != ThreadState.Running)
                        {
                            int activeTemp = handle.ActiveState;
                            int newActiveState = 2;
                            if (Interlocked.CompareExchange(ref handle._ActiveState, newActiveState, activeTemp) == activeTemp)
                            {
                                DecrementActiveThreads();
                            }
                        }
                        else
                        {
                            handle.AutoEventSet();
                            break;
                        }
                    }
                }
                else
                {
                    break;
                }
            }
        }                    
    }
}